/*
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.analysys.presto.connector.hbase.meta;

import com.analysys.presto.connector.hbase.adaptive.AdaptiveInsertColumnChannel;
import com.analysys.presto.connector.hbase.adaptive.HColumn;
import com.analysys.presto.connector.hbase.adaptive.HMetadataManager;
import com.analysys.presto.connector.hbase.adaptive.HTableInfo;
import com.analysys.presto.connector.hbase.connection.HBaseClientManager;
import com.analysys.presto.connector.hbase.frame.HBaseConnectorId;
import com.analysys.presto.connector.hbase.optimization.StartEndKey;
import com.analysys.presto.connector.hbase.utils.Constant;
import com.analysys.presto.connector.hbase.utils.Utils;
import com.facebook.airlift.log.Logger;
import com.facebook.presto.spi.ColumnHandle;
import com.facebook.presto.spi.ColumnMetadata;
import com.facebook.presto.spi.ConnectorInsertTableHandle;
import com.facebook.presto.spi.ConnectorSession;
import com.facebook.presto.spi.ConnectorTableHandle;
import com.facebook.presto.spi.ConnectorTableLayout;
import com.facebook.presto.spi.ConnectorTableLayoutHandle;
import com.facebook.presto.spi.ConnectorTableLayoutResult;
import com.facebook.presto.spi.ConnectorTableMetadata;
import com.facebook.presto.spi.Constraint;
import com.facebook.presto.spi.PrestoException;
import com.facebook.presto.spi.SchemaTableName;
import com.facebook.presto.spi.SchemaTablePrefix;
import com.facebook.presto.spi.TableNotFoundException;
import com.facebook.presto.spi.connector.ConnectorMetadata;
import com.facebook.presto.spi.connector.ConnectorOutputMetadata;
import com.facebook.presto.spi.statistics.ComputedStatistics;
import com.facebook.presto.spi.type.Type;
import com.facebook.presto.spi.type.VarcharType;
import com.google.common.base.Strings;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import io.airlift.slice.Slice;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import javax.inject.Inject;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.regex.Pattern;

import static com.analysys.presto.connector.hbase.utils.Constant.CONNECTOR_NAME;
import static com.analysys.presto.connector.hbase.utils.Types.checkType;
import static com.facebook.presto.spi.StandardErrorCode.ALREADY_EXISTS;
import static com.facebook.presto.spi.StandardErrorCode.HBASE_NOT_FORMAT;
import static com.facebook.presto.spi.StandardErrorCode.HBASE_TYPE_ERROR;
import static com.facebook.presto.spi.StandardErrorCode.SCHEMA_NOT_EMPTY;
import static com.facebook.presto.spi.adaptive.AdaptiveConstant.TABLE_PROPERTY_ROWKEYFORMAT;
import static com.facebook.presto.spi.adaptive.AdaptiveConstant.TABLE_PROPERTY_SEPARATOR;
import static com.google.common.base.Preconditions.checkArgument;
import static java.util.Objects.requireNonNull;

/**
 * HBase metadata
 *
 * @author wupeng
 * @date 2019/01/29
 */

public class HBaseMetadata
        implements ConnectorMetadata
{
    private static final Logger log = Logger.get(HBaseMetadata.class);

    private final HBaseConnectorId connectorId;
    private final HBaseTables hbaseTables;
    private final HBaseClientManager hbaseClientManager;

    private String table_property_separator = TABLE_PROPERTY_SEPARATOR;

    // adaptive
    private HMetadataManager hMetadataManager;

    @Inject
    public HBaseMetadata(HBaseConnectorId connectorId, HBaseTables hbaseTables, HBaseClientManager hbaseClientManager)
    {
        this.connectorId = connectorId;
        this.hbaseTables = requireNonNull(hbaseTables, "hbaseTables is null");
        this.hbaseClientManager = hbaseClientManager;

        // adptive
        this.hMetadataManager = hbaseClientManager.gethMetadataManager();
    }

    @Override
    public List<String> listSchemaNames(ConnectorSession connectorSession)
    {
        return listSchemaNames();
    }

    @Override
    public ConnectorTableHandle getTableHandle(ConnectorSession connectorSession, SchemaTableName schemaTableName)
    {
        requireNonNull(schemaTableName, "schemaTableName is null");
        Admin admin = null;
        ConnectorTableHandle connectorTableHandle;
        try {
            admin = hbaseClientManager.getAdmin();
            connectorTableHandle = hbaseTables.getTables(
                    admin, schemaTableName.getSchemaName()).get(schemaTableName);
        }
        finally {
            if (admin != null) {
                hbaseClientManager.close(admin);
            }
        }
        return connectorTableHandle;
    }

    @Override
    public List<ConnectorTableLayoutResult> getTableLayouts(ConnectorSession session, ConnectorTableHandle table, Constraint<ColumnHandle> constraint, Optional<Set<ColumnHandle>> desiredColumns)
    {
        HBaseTableHandle hBaseTableHandle = (HBaseTableHandle) table;
        ConnectorTableLayout layout = new ConnectorTableLayout(new HBaseTableLayoutHandle(hBaseTableHandle, constraint.getSummary(), StartEndKey.ofNull()));
        return ImmutableList.of(new ConnectorTableLayoutResult(layout, constraint.getSummary()));
    }

    @Override
    public ConnectorTableLayout getTableLayout(ConnectorSession session, ConnectorTableLayoutHandle handle)
    {
        return new ConnectorTableLayout(handle);
    }

    @Override
    public ConnectorTableMetadata getTableMetadata(ConnectorSession connectorSession,
            ConnectorTableHandle connectorTableHandle)
    {
        HBaseTableHandle hBaseTableHandle = (HBaseTableHandle) connectorTableHandle;
        SchemaTableName tableName = new SchemaTableName(hBaseTableHandle.getSchemaTableName().getSchemaName(),
                hBaseTableHandle.getSchemaTableName().getTableName());
        return this.getTableMetadata(tableName);
    }

    private ConnectorTableMetadata getTableMetadata(SchemaTableName tableName)
    {
        if (!this.listSchemaNames().contains(tableName.getSchemaName())) {
            return null;
        }
        else {
            HBaseTable table = hbaseClientManager.getTable(tableName.getSchemaName(), tableName.getTableName());
            return table == null ? null : new ConnectorTableMetadata(tableName, table.getColumnsMetadata());
        }
    }

    private List<String> listSchemaNames()
    {
        return ImmutableList.copyOf(hbaseTables.getSchemaNames());
    }

    @Override
    public List<SchemaTableName> listTables(ConnectorSession connectorSession, Optional<String> schemaName)
    {
        Admin admin = hbaseClientManager.getAdmin();
        List<SchemaTableName> schemaTableNames;
        try {
            schemaTableNames = new ArrayList<>(hbaseTables.getTables(admin, schemaName.orElse("")).keySet());
        }
        finally {
            if (admin != null) {
                hbaseClientManager.close(admin);
            }
        }
        return schemaTableNames;
    }

    @Override
    public Map<String, ColumnHandle> getColumnHandles(ConnectorSession connectorSession,
            ConnectorTableHandle connectorTableHandle)
    {
        HBaseTableHandle hBaseTableHandle = (HBaseTableHandle) connectorTableHandle;
        HBaseTable table = hbaseClientManager.getTable(hBaseTableHandle.getSchemaTableName().getSchemaName(),
                hBaseTableHandle.getSchemaTableName().getTableName());
        if (table == null) {
            throw new TableNotFoundException(hBaseTableHandle.getSchemaTableName());
        }
        else {
            ImmutableMap.Builder<String, ColumnHandle> columnHandles = ImmutableMap.builder();
            int index = 0;
            for (Iterator itr = table.getColumnsMetadata().iterator(); itr.hasNext(); ++index) {
                HBaseColumnMetadata column = (HBaseColumnMetadata) itr.next();
                columnHandles.put(column.getName(),
                        new HBaseColumnHandle(
                                connectorId.getId(), column.getFamily(), column.getName(),
                                column.getType(), index, column.isRowKey()));
            }
            return columnHandles.build();
        }
    }

    @Override
    public ColumnMetadata getColumnMetadata(ConnectorSession connectorSession,
            ConnectorTableHandle connectorTableHandle,
            ColumnHandle columnHandle)
    {
        checkType(connectorTableHandle, HBaseTableHandle.class, "tableHandle");
        return checkType(columnHandle, HBaseColumnHandle.class, "columnHandle").toColumnMetadata();
    }

    @Override
    public Map<SchemaTableName, List<ColumnMetadata>> listTableColumns(ConnectorSession connectorSession,
            SchemaTablePrefix schemaTablePrefix)
    {
        requireNonNull(schemaTablePrefix, "prefix is null");
        ImmutableMap.Builder<SchemaTableName, List<ColumnMetadata>> columns = ImmutableMap.builder();
        List<SchemaTableName> tables = this.listTables(connectorSession, schemaTablePrefix);
        for (SchemaTableName tableName : tables) {
            ConnectorTableMetadata tableMetadata = this.getTableMetadata(tableName);
            if (tableMetadata != null) {
                columns.put(tableName, tableMetadata.getColumns());
            }
        }
        return columns.build();
    }

    private List<SchemaTableName> listTables(ConnectorSession session, SchemaTablePrefix prefix)
    {
        return (prefix.getSchemaName() != null ?
                this.listTables(session, prefix.getSchemaName()) :
                ImmutableList.of(new SchemaTableName("", prefix.getTableName())));
    }

    @Override
    public void dropTable(ConnectorSession session, ConnectorTableHandle tableHandle)
    {
        HBaseTableHandle handle = (HBaseTableHandle) tableHandle;
        String schema = handle.getSchemaTableName().getSchemaName();
        String tableName = handle.getSchemaTableName().getTableName();
        hbaseTables.dropTable(schema, tableName);
        hMetadataManager.removeATable(schema, tableName);
    }

    /**
     * create snapshot
     *
     * @param snapshotName snapshot name
     * @param admin        admin
     * @param schemaName   schema name
     * @param tableName    table name
     * @throws IOException io exception
     */
    public static void createSnapshot(String snapshotName,
            Admin admin,
            String schemaName,
            String tableName)
            throws IOException
    {
        long start = System.currentTimeMillis();
        String fullTableName;
        if (Constant.HBASE_NAMESPACE_DEFAULT.equals(schemaName)
                || "".equals(schemaName)) {
            fullTableName = tableName;
        }
        else {
            fullTableName = schemaName + ":" + tableName;
        }
        SnapshotDescription snapshot = new SnapshotDescription(snapshotName, TableName.valueOf(fullTableName), SnapshotType.FLUSH);
        admin.snapshot(snapshot);
        log.info("createSnapshot: create snapshot " + snapshotName
                + " used " + (System.currentTimeMillis() - start) + " mill seconds.");
    }

    /**
     * create schema
     *
     * @param schemaName schema name
     * @throws PrestoException schema exists
     * @author luo
     */
    @Override
    public void createSchema(ConnectorSession session, String schemaName, Map<String, Object> properties)
    {
        if (this.listSchemaNames().contains(schemaName)) {
            throw new PrestoException(ALREADY_EXISTS, "The schema already exists.");
        }
        else {
            hbaseClientManager.createSchema(schemaName);
        }
    }

    /**
     * drop schema
     *
     * @param schemaName schema name
     * @throws PrestoException schema exists
     * @author luo
     */
    @Override
    public void dropSchema(ConnectorSession session, String schemaName)
    {
        if (!listTables(session, Optional.of(schemaName)).isEmpty()) {
            throw new PrestoException(SCHEMA_NOT_EMPTY, "Schema not empty: " + schemaName);
        }
        hbaseClientManager.dropSchema(schemaName);
    }

    /**
     * create table
     *
     * @throws PrestoException HBASE_NO_ROWKEY  HBASE_NOT_FORMAT
     * @author luo
     */
    @Override
    public void createTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, boolean ignoreExisting)
    {
        String schemaName = tableMetadata.getTable().getSchemaName();
        String tableName = tableMetadata.getTable().getTableName();
        // check whether table exists.
        HTableInfo hTableInfo = hMetadataManager.getTable(schemaName, tableName);
        if (hTableInfo != null) {
            throw new PrestoException(ALREADY_EXISTS, "HBase table has already exists.");
        }

        // check whether rowkey-format property exists.
        if (hMetadataManager.isAdaptiveEnable()) {
            checkRowkeyFormat(tableMetadata);
        }
        // check whether rowkey exists.
        else {
            checkRowkeyExist(tableMetadata);
        }

        // create table in hbase and metadata manager.
        HTableInfo.Builder newTableBuilder = new HTableInfo.Builder(tableName)
                .setSchemaName(schemaName)
                .setRowKeyFormat((String) tableMetadata.getProperties().get(TABLE_PROPERTY_ROWKEYFORMAT));

        Set<String> cfSet = new HashSet<>();
        for (ColumnMetadata columnMetadata : tableMetadata.getColumns()) {
            String columnName = columnMetadata.getName();

            String columnType = columnMetadata.getType().toString();
            checkColumnType(columnType);

            String cf = (String) columnMetadata.getProperties().get("cf");
            cfSet.add(cf);

            if (!hMetadataManager.isAdaptiveEnable() && columnName.equals("rowkey")) {
                newTableBuilder.addRowKey(columnType);
            }
            else {
                newTableBuilder.addColumn(cf, columnName, columnType);
            }
        }

        // 1. add to hbase.
        hbaseClientManager.createTable(schemaName, tableName, cfSet);

        // 2. add to meta manager.
        HTableInfo newTable = newTableBuilder.build();
        hMetadataManager.addATable(newTable);
    }

    private void checkRowkeyExist(ConnectorTableMetadata tableMetadata)
    {
        boolean hasRowkey = tableMetadata.getColumns().stream().anyMatch(column -> column.getName().equals("rowkey"));
        if (!hasRowkey) {
            throw new PrestoException(HBASE_NOT_FORMAT, "HBase table has no rowkey.");
        }
    }

    private void checkRowkeyFormat(ConnectorTableMetadata tableMetadata)
    {
        String rowkeyFormat = (String) tableMetadata.getProperties().get(TABLE_PROPERTY_ROWKEYFORMAT);

        if (rowkeyFormat == null) {
            throw new PrestoException(HBASE_NOT_FORMAT, "HBase table has not rowkey-format.");
        }

        // pattern is salt|%s-%d|%s-%d
        String pattern = "^salt"
                + table_property_separator + "[\\s\\S]*\\-[0-9]+"
                + table_property_separator + "[\\s\\S]*\\-[0-9]+$";
        boolean isMatch = Pattern.matches(pattern, rowkeyFormat);

        // not match
        if (!isMatch) {
            throw new PrestoException(HBASE_NOT_FORMAT, "HBase rowkey-format is not valid, check whether rowkey-format is salt|%s-%d|%s-%d.");
        }

        String indexColumnName = (rowkeyFormat.split(table_property_separator)[1]).split("-")[0];
        String primaryKeyColumnName = (rowkeyFormat.split(table_property_separator)[2]).split("-")[0];

        // check whether indexColumn exists.
        boolean hasIndexColumn = tableMetadata.getColumns().stream().anyMatch(column -> column.getName().equals(indexColumnName));
        if (!hasIndexColumn) {
            throw new PrestoException(HBASE_NOT_FORMAT, "HBase rowkey-format is not valid, column: " + indexColumnName + " not exists.");
        }

        // check whether pkColumn exists.
        boolean hasPKColumn = tableMetadata.getColumns().stream().anyMatch(column -> column.getName().equals(primaryKeyColumnName));
        if (!hasPKColumn) {
            throw new PrestoException(HBASE_NOT_FORMAT, "HBase rowkey-format is not valid, column: " + primaryKeyColumnName + " not exists.");
        }
    }

    /**
     * Check whether column type is valid, now only support varchar, int, bigint, double.
     *
     * @author luo
     */
    private void checkColumnType(String columnType)
    {
        if (columnType.equals("integer")) {
            return;
        }
        if (columnType.equals("bigint")) {
            return;
        }
        if (columnType.equals("double")) {
            return;
        }
        if (columnType.equals("varchar")) {
            return;
        }
        throw new PrestoException(HBASE_TYPE_ERROR, "HBase column type error, only support int/bigint/varchar/double.");
    }

    public void loadData(String schemaName, String tableName, String fileURL) {
        HTableInfo hTableInfo = hMetadataManager.getTable(schemaName, tableName);
        List<HColumn> hColumns = hTableInfo.gethColumnList();
        int csvBatchSize = hbaseClientManager.getConfig().getCsvBatchSize();
        try {
            Connection connection = hbaseClientManager.createConnection();
            Table table = connection.getTable(TableName.valueOf(schemaName + ":" + tableName));
            String indexColumnName = hTableInfo.getIndexColumnName();
            int indexColumnLength = hTableInfo.getIndexColumnLength();
            String pkColumnName = hTableInfo.getPkColumnName();
            int pkColumnLength = hTableInfo.getPkColumnLength();
            int idxPos = 0, pkPos = 0;
            int cnt = 0;
            for (HColumn hColumn : hColumns) {
                if (hColumn.getColumnName().equals(indexColumnName)) idxPos = cnt;
                if (hColumn.getColumnName().equals(pkColumnName)) pkPos = cnt;
                cnt++;
            }

            File csv = new File(fileURL);
            BufferedReader br = null;
            br = new BufferedReader(new FileReader(csv));
            String line = "";
            List<Put> puts = new ArrayList<>();
            while ((line = br.readLine()) != null) {
                Put put = insertForLoad(idxPos, indexColumnLength, pkPos, pkColumnLength, line, hColumns);
                puts.add(put);
                if (puts.size() == csvBatchSize) {
                    table.put(puts);
                    puts.clear();
                }
            }
            if (!puts.isEmpty()) table.put(puts);
            connection.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    private Put insertForLoad(int idxPos, int indexColumnLength, int pkPos, int pkColumnLength, String line, List<HColumn> hColumns) {
        String[] split = line.split(hbaseClientManager.getConfig().getCsvSplit());
        String rowKey = joinIndexAndPKColumn(split[idxPos], indexColumnLength, split[pkPos], pkColumnLength);
        Put put = new Put(Bytes.toBytes(rowKey));
        int cnt = 0;
        for (String item : split) {
            String family = hColumns.get(cnt).getFamily();
            String columnName = hColumns.get(cnt).getColumnName();
            put.addColumn(Bytes.toBytes(family), Bytes.toBytes(columnName), Bytes.toBytes(item));
            cnt++;
        }
        return put;
    }

    private String joinIndexAndPKColumn(String idxVal, int indexColumnLength, String pkVal, int pkColumnLength) {
        StringBuilder stringBuilder = new StringBuilder();
        stringBuilder.append(createSaltStr());
        stringBuilder.append(createQualifiedColumnStr(idxVal, indexColumnLength));
        stringBuilder.append(createQualifiedColumnStr(pkVal, pkColumnLength));
        return stringBuilder.toString();
    }

    private String createQualifiedColumnStr(String val, int length) {
        String qualifiedColumnStr = fillStart(val, length, '0');
        return qualifiedColumnStr;
    }

    /**
     * fill the string
     * fillStart("ab", 5, '2')  =>  "222ab"
     */
    private String fillStart(String str, int len, char ch) {
        return Strings.padStart(str, len, ch);
    }

    private String createSaltStr() {
        int max = 9, min = 0;
        int ran = (int) (Math.random() * (max - min) + min);
        return Integer.toString(ran);
    }
    // ----------------------------------- start insert -----------------------------------
    @Override
    public ConnectorInsertTableHandle beginInsert(ConnectorSession session, ConnectorTableHandle connectorTableHandle)
    {
        HBaseTableHandle tableHandle = fromConnectorTableHandle(connectorTableHandle);
        String schemaName = tableHandle.getSchemaTableName().getSchemaName();
        String tableName = tableHandle.getSchemaTableName().getTableName();
        try {
            // adaptive
            HTableInfo hTableInfo = hMetadataManager.getTable(schemaName, tableName);
            requireNonNull(hTableInfo, String.format("The metadata of table %s.%s is null", schemaName, tableName));
            List<HColumn> hColumns = hTableInfo.gethColumnList();

            List<String> columnNames = new ArrayList<>(hColumns.size());
            List<Type> columnTypes = new ArrayList<>(hColumns.size());
            Map<String, String> colNameAndFamilyNameMap = new HashMap<>();
            for (HColumn hColumn : hColumns) {
                columnNames.add(hColumn.getColumnName());
                columnTypes.add(Utils.matchType(hColumn.getType()));
                colNameAndFamilyNameMap.put(hColumn.getColumnName(), hColumn.getFamily());
            }
            int rowKeyColumnChannel = this.findRowKeyChannel(hTableInfo.gethColumnList());
            if (!this.hMetadataManager.isAdaptiveEnable()) {
                return new HBaseInsertTableHandle(
                        connectorId.getId(),
                        tableHandle.getSchemaTableName(),
                        columnNames,
                        columnTypes,
                        rowKeyColumnChannel,
                        colNameAndFamilyNameMap,
                        null);
            }
            else {
                String indexColumnName = hTableInfo.getIndexColumnName();
                String pkColumnName = hTableInfo.getPkColumnName();
                int indexColumnLength = hTableInfo.getIndexColumnLength();
                int pkColumnLength = hTableInfo.getPkColumnLength();

                int indexColumnChannel = this.findChannel(hTableInfo.gethColumnList(), indexColumnName);
                int pkColumnChannel = this.findChannel(hTableInfo.gethColumnList(), pkColumnName);

                return new HBaseInsertTableHandle(
                        connectorId.getId(),
                        tableHandle.getSchemaTableName(),
                        columnNames,
                        columnTypes,
                        rowKeyColumnChannel,
                        colNameAndFamilyNameMap,
                        new AdaptiveInsertColumnChannel(indexColumnLength, indexColumnChannel, pkColumnLength, pkColumnChannel));
            }

        }
        catch (Exception ex) {
            log.error(ex.getMessage(), ex);
        }
        return null;
    }

    private int findChannel(List<HColumn> hColumns, String column)
    {
        for (int i = 0; i < hColumns.size(); i++) {
            if (hColumns.get(i).getColumnName().equals(column)) {
                return i;
            }
        }
        return -1;
    }

    private int findRowKeyChannel(List<HColumn> columns)
    {
        for (int i = 0; i < columns.size(); i++) {
            if (columns.get(i).isRowKey()) {
                return i;
            }
        }
        // Table must specify rowKey column's name.
        // So it's impossible to be here.
        return -1;
    }

    @Override
    public Optional<ConnectorOutputMetadata> finishInsert(ConnectorSession session,
            ConnectorInsertTableHandle insertHandle,
            Collection<Slice> fragments,
            Collection<ComputedStatistics> computedStatistics)
    {
        return Optional.empty();
    }

    private HBaseTableHandle fromConnectorTableHandle(ConnectorTableHandle tableHandle)
    {
        return checkType(tableHandle, HBaseTableHandle.class, "tableHandle");
    }
    // ----------------------------------- end insert -----------------------------------

    // --------------- support delete function start ---------------
    @Override
    public ColumnHandle getUpdateRowIdColumnHandle(ConnectorSession session, ConnectorTableHandle tableHandle)
    {
        HBaseTableHandle hth = (HBaseTableHandle) tableHandle;
        String schemaName = hth.getSchemaTableName().getSchemaName();
        String tableName = hth.getSchemaTableName().getTableName();

        HTableInfo hTableInfo = hMetadataManager.getTable(schemaName, tableName);

        requireNonNull(hTableInfo, String.format("Table %s.%s has no metadata, please check zkPath:%s, zkNode:%s",
                schemaName,
                tableName,
                hbaseClientManager.getConfig().getMetadataZKHost(),
                hbaseClientManager.getConfig().getMetadataZKNode()));

        Optional<HColumn> rowKeyOpt = hTableInfo.gethColumnList().stream().filter(HColumn::isRowKey).findFirst();
        checkArgument(rowKeyOpt.isPresent(), String.format("Table %s.%s has no rowKey! Please check zkPath:%s, zkNode:%s",
                schemaName,
                tableName,
                hbaseClientManager.getConfig().getMetadataZKHost(),
                hbaseClientManager.getConfig().getMetadataZKNode()));

        HColumn rowKeyInfo = rowKeyOpt.get();
        // HBaseColumnHandle's attributes cannot be all the same with the REAL rowKey column,
        // Or there will be a java.lang.IllegalArgumentException: Multiple entries with same value Exception.
        return new HBaseColumnHandle(CONNECTOR_NAME, "",
                rowKeyInfo.getColumnName(), VarcharType.VARCHAR,
                hTableInfo.gethColumnList().indexOf(rowKeyOpt.get()),
                rowKeyInfo.isRowKey());
    }

    @Override
    public ConnectorTableHandle beginDelete(ConnectorSession session, ConnectorTableHandle tableHandle)
    {
        return fromConnectorTableHandle(tableHandle);
    }

    @Override
    public void finishDelete(ConnectorSession session, ConnectorTableHandle tableHandle, Collection<Slice> fragments)
    {
    }

////     --------------- support delete function end ---------------
}
